package com.atguigu.gmall.realtime.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.gmall.realtime.app.func.DimAsyncFunction;
import com.atguigu.gmall.realtime.beans.TradeSkuOrderBean;
import com.atguigu.gmall.realtime.utils.DateFormatUtil;
import com.atguigu.gmall.realtime.utils.MyClickhouseUtil;
import com.atguigu.gmall.realtime.utils.MyKafkaUtil;
import com.atguigu.gmall.realtime.utils.TimestampLtz3CompareUtil;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.common.state.StateTtlConfig;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.*;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.util.Collector;

import java.time.Duration;
import java.util.Collections;
import java.util.HashSet;
import java.util.concurrent.TimeUnit;

/**
 * @author Felix
 * @date 2022/12/17
 * 交易域：sku粒度下单聚合统计
 * 维度：sku
 * 度量：独立用户数、订单数、原始金额、活动减免金额、优惠券减免金额和订单金额
 * 需要启动的进程
 *      zk、kafka、maxwell、hdfs、hbase、Redis、clickhouse
 *      DwdTradeOrderPreProcess、DwdTradeOrderDetail、DwsTradeSkuOrderWindow
 * 开发流程：
 *      基本环境准备
 *      检查点相关的设置
 *      从kafka的下单主题中读取下单数据
 *      对读取的数据类型进行转换    jsonStr->jsonObj
 *      去重
 *          为什么会产生重复数据？
 *              我们的数据来源是下单主题，下单主题的数据是从订单预处理中获取的，订单预处理是将订单、订单明细、明细活动、明细优惠券、字典表五张表进行关联
 *              和明细活动以及优惠券在关联的时候，使用的是左外连接，如果左表数据先来，右表数后到，会产生如下数据
 *              +I  左表  null
 *              -D  左表  null
 *              +I  左表  右表
 *              将上面的数据写到kafka主题的话，kafka主题会接收到3条消息
 *              左表  null
 *              null
 *              左表  右表
 *              我们在从预处理主题中读取数据的时候，使用的是kafka连接器，会将null消息过滤掉，剩下的两条消息是重复，需要进行去重
 *          去重思路：
 *              Flink的状态编程 + 定时器
 *              按照唯一键（订单明细的id进行分组）
 *              对分组之后的数据进行处理
 *                  当第一条数据来的时候，先判断状态中是否有数据，如果没有，将这条数据放到状态中，并且注册一个5秒之后执行的定时器
 *                  当第二条数据来的时候，会和状态中已经存在的数据进行时间的比较，将时间大的数据放到状态中
 *                  等到定时器被触发执行的时候，将状态中的数据向下游传递
 *      类型转换    jsonObj->实体类对象
 *      按照用户id进行分组
 *      判断是否下单独立用户
 *      指定Watermark以及提取事件时间字段
 *      按照skuId分组
 *      开窗
 *      聚合计算
 *      维度关联
 *          基本实现
 *              PhoenixUtil
 *                  List<T> queryList(String sql,Class<T> clz)
 *              DimUtil
 *                  JSONObject getDimInfoNoCache(String tableName,Tuple2<String,String> ... params)
 *          优化1：旁路缓存
 *              思路：先从缓存中获取维度数据，如果从缓存中能够获取到要关联的维度(缓存命中),直接将缓存的缓存的维度作为返回值返回；
 *                  如果从缓存中没有获取到要关联的维度，发送请求到Phoenix表中查询维度，并且将查询的结果放到缓存中，方便下次查询。
 *              缓存产品选择：
 *                  状态：     性能高     维护性差
 *                  Redis：   性能也不差  维护性好    √
 *              Redis相关的设置：
 *                  key:    dim:维度表名:主键1_主键2
 *                  type:   String
 *                  expire: 1day
 *                  注意：   如果Phoenix维度表发生了变化，需要将Redis中缓存的数据 清除掉
 *          优化2：异步IO
 *              为什么要使用异步IO？
 *                  要想提升和外部系统交互能力，可以加大并行度，但是加大并行度，意味着更多的硬件资源，不可能无限制的提升。
 *                  所以我们要考虑在硬件资源有限的情况下，提升处理能力
 *                  默认情况下，如果使用map算子对流中数据进行处理，使用的是同步的方式
 *                  和外部系统交互的时候，可以通过Flink提供的API，发送异步请求，对流中数据进行处理，提升处理能力
 *              请求方式
 *                  AsyncDataStream.[un]orderedWait(
 *                      流，
 *                      异步请求实现方式    implements AsyncFunction,
 *                      超时时间，
 *                      时间单位
 *                  )
 *              封装
 *                  abstract class DimAsyncFunction extends RichAsyncFunction{
 *                      open{
 *                          初始化线程池
 *                      }
 *                      asyncInvoke{
 *                          //启动线程
 *                          //根据流中的对象获取要关联的维度主键（抽象）
 *                          //根据维度的主键   到维度表中获取对应的维度对象
 *                          //将维度对象的属性补充到流中对象上（抽象）
 *                          //将补充完维度属性的的对象向下游传递
 *                      }
 *                  }
 *             模板方法设计模式：在父类中定义完成某一个功能的核心算法骨架(步骤)，某些具体的实现延迟到子类中去完成，在不改变父类核心算法骨架的前提下，
 *                          每一个子类都可以有自己不同的实现。
 *      将关联维度后的数据写到CK中
 */
public class DwsTradeSkuOrderWindow {
    public static void main(String[] args) throws Exception {
        //TODO 1.基本环境准备
        //1.1 指定流处理环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //1.2 设置并行度
        env.setParallelism(4);
        //TODO 2.检查点相关的设置(略)
        //TODO 3.从kakfa主题中读取下单数据
        //3.1 声明消费的主题以及消费者组
        String topic = "dwd_trade_order_detail";
        String groupId = "dws_trade_sku_order_group";
        //3.2 创建消费者对象
        FlinkKafkaConsumer<String> kafkaConsumer = MyKafkaUtil.getKafkaConsumer(topic, groupId);
        //3.3 消费数据  封装为流
        DataStreamSource<String> kafkaStrDS = env.addSource(kafkaConsumer);

        //TODO 4.对流中数据类型进行转换            jsonStr->jsonObj
        SingleOutputStreamOperator<JSONObject> jsonObjDS = kafkaStrDS.map(JSON::parseObject);
        // {"create_time":"2022-12-03 09:24:40","sku_num":"3","activity_rule_id":"5","split_original_amount":"387.0000",
        // "sku_id":"26","date_id":"2022-12-03","source_type_name":"用户查询","user_id":"165","province_id":"18",
        // "source_type_code":"2401","row_op_ts":"2022-12-17 01:24:40.915Z","activity_id":"3","sku_name":"索芙特",
        // "id":"259","order_id":"115","split_activity_amount":"39.0","split_total_amount":"348.0","ts":"1671240280"}
        // jsonObjDS.print(">>>");

        //TODO 5.按照唯一键(订单明细id)进行分组
        KeyedStream<JSONObject, String> orderDetailIdKeyedDS = jsonObjDS.keyBy(jsonObj -> jsonObj.getString("id"));
        //TODO 6.去重     状态编程 + 定时器
        SingleOutputStreamOperator<JSONObject> distinctDS = orderDetailIdKeyedDS.process(
            new KeyedProcessFunction<String, JSONObject, JSONObject>() {
                private ValueState<JSONObject> lastValueState;

                @Override
                public void open(Configuration parameters) throws Exception {
                    lastValueState
                        = getRuntimeContext().getState(new ValueStateDescriptor<JSONObject>("lastValueState",JSONObject.class));
                }

                @Override
                public void processElement(JSONObject jsonObj, Context ctx, Collector<JSONObject> out) throws Exception {
                    //从状态中获取数据
                    JSONObject lastValue = lastValueState.value();
                    if(lastValue == null){
                        //状态中还没有存放数据，说明现在处理的数据是当前订单明细对应的第一条数据，将这条数据放到状态中,并注册一个定时器
                        lastValueState.update(jsonObj);
                        long currentProcessingTime = ctx.timerService().currentProcessingTime();
                        ctx.timerService().registerProcessingTimeTimer(currentProcessingTime + 5000L);
                    }else{
                        //说明当前订单明细已经有对应的数据进来并放到状态中，用当前处理的数据和状态中存放的数据进行时间的比较
                        //"row_op_ts":"2022-12-17 01:24:40.915Z"
                        //"row_op_ts":"2022-12-17 01:24:40.91Z"
                        String lastRowOpTs = lastValue.getString("row_op_ts");
                        String curRowOpTs = jsonObj.getString("row_op_ts");
                        if(TimestampLtz3CompareUtil.compare(lastRowOpTs,curRowOpTs)<=0){
                            lastValueState.update(jsonObj);
                        }
                    }
                }

                @Override
                public void onTimer(long timestamp, OnTimerContext ctx, Collector<JSONObject> out) throws Exception {
                    JSONObject value = lastValueState.value();
                    if(value != null){
                        out.collect(value);
                    }
                    lastValueState.clear();
                }
            }
        );
        //TODO 7.对去重后流中的数据进行类型转换       jsonObj->实体类对象
        SingleOutputStreamOperator<TradeSkuOrderBean> orderBeanDS = distinctDS.map(
            new MapFunction<JSONObject, TradeSkuOrderBean>() {
                @Override
                public TradeSkuOrderBean map(JSONObject jsonObj) throws Exception {
                    // {"create_time":"2022-12-03 09:24:40","sku_num":"3","activity_rule_id":"5","split_original_amount":"387.0000",
                    // "sku_id":"26","date_id":"2022-12-03","source_type_name":"用户查询","user_id":"165","province_id":"18",
                    // "source_type_code":"2401","row_op_ts":"2022-12-17 01:24:40.915Z","activity_id":"3","sku_name":"索芙特",
                    // "id":"259","order_id":"115","split_activity_amount":"39.0","split_total_amount":"348.0","ts":"1671240280"}
                    String skuId = jsonObj.getString("sku_id");
                    String userId = jsonObj.getString("user_id");
                    String orderId = jsonObj.getString("order_id");
                    Double splitOriginalAmount = jsonObj.getDouble("split_original_amount");
                    Double splitActivityAmount = jsonObj.getDouble("split_activity_amount");
                    Double splitCouponAmount = jsonObj.getDouble("split_coupon_amount");
                    Double splitTotalAmount = jsonObj.getDouble("split_total_amount");
                    Long ts = jsonObj.getLong("ts")*1000;


                    TradeSkuOrderBean orderBean = TradeSkuOrderBean.builder()
                        .skuId(skuId)
                        .userId(userId)
                        .orderUuCount(0L)
                        .orderIdSet(new HashSet<>(Collections.singleton(orderId)))
                        .originalAmount(splitOriginalAmount)
                        .activityAmount(splitActivityAmount == null ? 0.0 : splitActivityAmount)
                        .couponAmount(splitCouponAmount == null ? 0.0 : splitCouponAmount)
                        .orderAmount(splitTotalAmount)
                        .ts(ts)
                        .build();
                    return orderBean;
                }
            }
        );

        //TODO 8.按照用户id进行分组
        KeyedStream<TradeSkuOrderBean, String> userIdKeyedDS = orderBeanDS.keyBy(TradeSkuOrderBean::getUserId);
        //TODO 9.使用Flink的状态编程  判断是否为下单独立用户
        SingleOutputStreamOperator<TradeSkuOrderBean> orderUUDS = userIdKeyedDS.map(
            new RichMapFunction<TradeSkuOrderBean, TradeSkuOrderBean>() {
                private ValueState<String> lastOrderDateState;

                @Override
                public void open(Configuration parameters) throws Exception {
                    ValueStateDescriptor<String> valueStateDescriptor
                        = new ValueStateDescriptor<String>("lastOrderDateState", String.class);
                    valueStateDescriptor.enableTimeToLive(StateTtlConfig.newBuilder(Time.days(1)).build());
                    lastOrderDateState = getRuntimeContext().getState(valueStateDescriptor);

                }

                @Override
                public TradeSkuOrderBean map(TradeSkuOrderBean orderBean) throws Exception {
                    String lastOrderDate = lastOrderDateState.value();
                    Long ts = orderBean.getTs();
                    String curOrderDate = DateFormatUtil.toDate(ts);
                    if (StringUtils.isEmpty(lastOrderDate) || !lastOrderDate.equals(curOrderDate)) {
                        orderBean.setOrderUuCount(1L);
                        lastOrderDateState.update(curOrderDate);
                    }
                    return orderBean;
                }
            }
        );
        //TODO 10.指定Watermark以及提取事件时间字段
        SingleOutputStreamOperator<TradeSkuOrderBean> withWatermarkDS = orderUUDS.assignTimestampsAndWatermarks(
            WatermarkStrategy
                .<TradeSkuOrderBean>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                .withTimestampAssigner(
                    new SerializableTimestampAssigner<TradeSkuOrderBean>() {
                        @Override
                        public long extractTimestamp(TradeSkuOrderBean orderBean, long recordTimestamp) {
                            return orderBean.getTs();
                        }
                    }
                )
        );
        //TODO 11.按照统计的维度进行分组
        KeyedStream<TradeSkuOrderBean, String> skuIdKeyedDS = withWatermarkDS.keyBy(TradeSkuOrderBean::getSkuId);

        //TODO 12.开窗
        WindowedStream<TradeSkuOrderBean, String, TimeWindow> windowDS
            = skuIdKeyedDS.window(TumblingEventTimeWindows.of(org.apache.flink.streaming.api.windowing.time.Time.seconds(10)));

        //TODO 13.聚合
        SingleOutputStreamOperator<TradeSkuOrderBean> reduceDS = windowDS.reduce(
            new ReduceFunction<TradeSkuOrderBean>() {
                @Override
                public TradeSkuOrderBean reduce(TradeSkuOrderBean value1, TradeSkuOrderBean value2) throws Exception {
                    value1.setOrderUuCount(value1.getOrderUuCount() + value2.getOrderUuCount());
                    value1.getOrderIdSet().addAll(value2.getOrderIdSet());
                    value1.setOriginalAmount(value1.getOriginalAmount() + value2.getOriginalAmount());
                    value1.setActivityAmount(value1.getActivityAmount() + value2.getActivityAmount());
                    value1.setCouponAmount(value1.getCouponAmount() + value2.getCouponAmount());
                    value1.setOrderAmount(value1.getOrderAmount() + value2.getOrderAmount());
                    return value1;
                }
            },
            new WindowFunction<TradeSkuOrderBean, TradeSkuOrderBean, String, TimeWindow>() {
                @Override
                public void apply(String groupId, TimeWindow window, Iterable<TradeSkuOrderBean> input, Collector<TradeSkuOrderBean> out) throws Exception {
                    String stt = DateFormatUtil.toYmdHms(window.getStart());
                    String edt = DateFormatUtil.toYmdHms(window.getEnd());

                    for (TradeSkuOrderBean orderBean : input) {
                        orderBean.setStt(stt);
                        orderBean.setEdt(edt);
                        orderBean.setTs(System.currentTimeMillis());
                        orderBean.setOrderCount((long) orderBean.getOrderIdSet().size());
                        out.collect(orderBean);
                    }
                }
            }
        );
        // reduceDS.print(">>>>");
        //TODO 14.关联商品sku维度
        /*
        //使用map算子对流中数据进行处理，使用的是同步的处理方式
        SingleOutputStreamOperator<TradeSkuOrderBean> withSkuInfoDS = reduceDS.map(new MapFunction<TradeSkuOrderBean, TradeSkuOrderBean>() {
            @Override
            public TradeSkuOrderBean map(TradeSkuOrderBean orderBean) throws Exception {
                //根据流中的对象  获取要关联的维度的主键
                String skuId = orderBean.getSkuId();
                //根据维度的主键到维度表中获取对应的维度数据
                //ID,SPU_ID,PRICE,SKU_NAME,SKU_DESC,WEIGHT,TM_ID,CATEGORY3_ID,SKU_DEFAULT_IMG,IS_SALE,CREATE_TIME
                JSONObject dimInfoJsonObj = DimUtil.getDimInfo("dim_sku_info",skuId);
                //将维度对象相关的维度属性  补充到流中的对象上
                orderBean.setSkuName(dimInfoJsonObj.getString("SKU_NAME"));
                orderBean.setSpuId(dimInfoJsonObj.getString("SPU_ID"));
                orderBean.setTrademarkId(dimInfoJsonObj.getString("TM_ID"));
                orderBean.setCategory3Id(dimInfoJsonObj.getString("CATEGORY3_ID"));
                return orderBean;
            }
        });

        withSkuInfoDS.print(">>>");*/
        // 将异步I/O操作应用于DataStream作为DataStream的一次转换操作
        SingleOutputStreamOperator<TradeSkuOrderBean> withSkuInfoDS = AsyncDataStream.unorderedWait(
            reduceDS,
            //如何发送异步请求  实现分发请求的 AsyncFunction
            new DimAsyncFunction<TradeSkuOrderBean>("dim_sku_info") {
                @Override
                public void join(TradeSkuOrderBean orderBean, JSONObject dimInfoJsonObj) {
                    orderBean.setSkuName(dimInfoJsonObj.getString("SKU_NAME"));
                    orderBean.setSpuId(dimInfoJsonObj.getString("SPU_ID"));
                    orderBean.setTrademarkId(dimInfoJsonObj.getString("TM_ID"));
                    orderBean.setCategory3Id(dimInfoJsonObj.getString("CATEGORY3_ID"));
                }

                @Override
                public String getKey(TradeSkuOrderBean orderBean) {
                    return orderBean.getSkuId();
                }
            },
            60,
            TimeUnit.SECONDS
        );


        //TODO 15.关联商品spu维度
        SingleOutputStreamOperator<TradeSkuOrderBean> withSpuInfoDS = AsyncDataStream.unorderedWait(
            withSkuInfoDS,
            new DimAsyncFunction<TradeSkuOrderBean>("dim_spu_info") {
                @Override
                public void join(TradeSkuOrderBean orderBean, JSONObject dimInfoJsonObj) {
                    orderBean.setSpuName(dimInfoJsonObj.getString("SPU_NAME"));
                }

                @Override
                public String getKey(TradeSkuOrderBean orderBean) {
                    return orderBean.getSpuId();
                }
            },
            60, TimeUnit.SECONDS
        );
        //TODO 16.关联商品tm维度
        SingleOutputStreamOperator<TradeSkuOrderBean> withTmInfoDS = AsyncDataStream.unorderedWait(
            withSpuInfoDS,
            new DimAsyncFunction<TradeSkuOrderBean>("dim_base_trademark") {
                @Override
                public void join(TradeSkuOrderBean orderBean, JSONObject dimInfoJsonObj) {
                    orderBean.setTrademarkName(dimInfoJsonObj.getString("TM_NAME"));
                }

                @Override
                public String getKey(TradeSkuOrderBean orderBean) {
                    return orderBean.getTrademarkId();
                }
            },
            60, TimeUnit.SECONDS
        );
        // withTmInfoDS.print(">>>>");

        //TODO 17.关联商品category3维度
        SingleOutputStreamOperator<TradeSkuOrderBean> withCategory3DS = AsyncDataStream.unorderedWait(
            withTmInfoDS,
            new DimAsyncFunction<TradeSkuOrderBean>("dim_base_category3") {
                @Override
                public void join(TradeSkuOrderBean orderBean, JSONObject dimInfoJsonObj) {
                    // ID,NAME,CATEGORY2_ID
                    orderBean.setCategory3Name(dimInfoJsonObj.getString("NAME"));
                    orderBean.setCategory2Id(dimInfoJsonObj.getString("CATEGORY2_ID"));
                }

                @Override
                public String getKey(TradeSkuOrderBean orderBean) {
                    return orderBean.getCategory3Id();
                }
            },
            60, TimeUnit.SECONDS
        );
        //TODO 18.关联商品category2维度
        SingleOutputStreamOperator<TradeSkuOrderBean> withCategory2DS = AsyncDataStream.unorderedWait(
            withCategory3DS,
            new DimAsyncFunction<TradeSkuOrderBean>("dim_base_category2") {
                @Override
                public void join(TradeSkuOrderBean orderBean, JSONObject dimInfoJsonObj) {
                    // ID,NAME,CATEGORY1_ID
                    orderBean.setCategory2Name(dimInfoJsonObj.getString("NAME"));
                    orderBean.setCategory1Id(dimInfoJsonObj.getString("CATEGORY1_ID"));
                }

                @Override
                public String getKey(TradeSkuOrderBean orderBean) {
                    return orderBean.getCategory2Id();
                }
            },
            60, TimeUnit.SECONDS
        );
        //TODO 19.关联商品category1维度
        SingleOutputStreamOperator<TradeSkuOrderBean> withCategory1DS = AsyncDataStream.unorderedWait(
            withCategory2DS,
            new DimAsyncFunction<TradeSkuOrderBean>("dim_base_category1") {
                @Override
                public void join(TradeSkuOrderBean orderBean, JSONObject dimInfoJsonObj) {
                    orderBean.setCategory1Name(dimInfoJsonObj.getString("NAME"));
                }

                @Override
                public String getKey(TradeSkuOrderBean orderBean) {
                    return orderBean.getCategory1Id();
                }
            },
            60, TimeUnit.SECONDS
        );
        //TODO 20.将关联结果写到CK中

        withCategory1DS.print(">>>");
        withCategory1DS.addSink(MyClickhouseUtil.getSinkFunction("insert into dws_trade_sku_order_window values(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)"));

        env.execute();
    }
}
